package hust.rpc;

import hust.rpc.handler.MessageDecoder;
import hust.rpc.handler.MessageEncoder;
import hust.rpc.handler.RpcInvokeHandler;
import hust.rpc.pojo.Host;
import hust.rpc.utils.CommonUtils;
import hust.rpc.utils.SingletonUtils;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

/**
 * Channel通道提供者
 * 用于消费者获取Channel对象(与提供者的连接)
 */
@Slf4j
public class ChannelProvider {

    // 复用的Bootstrap对象
    private static final Bootstrap bootstrap;

    static {
        // 初始化Bootstrap对象
        bootstrap = new Bootstrap();
        bootstrap.group(new NioEventLoopGroup())
                .channel(NioSocketChannel.class)
                // 关闭TCP的默认开启的Nagle算法
                .option(ChannelOption.TCP_NODELAY, true)
                // 开启TCP底层心跳机制
                .option(ChannelOption.SO_KEEPALIVE, true)
                // 设置连接超时时间
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000);
    }

    /**
     * 连接服务提供者
     *
     * @param host 服务提供者地址信息
     * @return Channel 与服务提供者的Channel通道
     */
    public static Channel connect(Host host) throws ExecutionException, InterruptedException {
        // 设置handler处理器
        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel socketChannel) throws Exception {
                ChannelPipeline pipeline = socketChannel.pipeline();
                pipeline.addLast(new MessageEncoder());
                pipeline.addLast(new MessageDecoder());
                pipeline.addLast(SingletonUtils.getInstance(RpcInvokeHandler.class));
            }
        });
        // 创建Future用于同步获取Channel
        CompletableFuture<Channel> future = new CompletableFuture<>();
        // 连接服务提供者
        // 并添加监听器
        bootstrap.connect(host.getIp(), host.getPort()).addListener((ChannelFutureListener) f -> {
            // 如果连接成功
            if (f.isSuccess()) {
                log.info("消费者连接提供者{}成功~~~", CommonUtils.channelToHost(f.channel()));
                // 回调设置Future结果,Future.get()返回
                future.complete(f.channel());
            } else {
                // TODO 统一异常
                throw new RuntimeException("连接提供者失败~~~");
            }
        });
        // 阻塞直到成功连接服务提供者,返回对应的Channel
        return future.get();
    }
}
